package com.wugui.datatx.core.thread;

import com.wugui.datatx.core.biz.model.HandleCallbackParam;
import com.wugui.datatx.core.biz.model.ReturnT;
import com.wugui.datatx.core.biz.model.TriggerParam;
import com.wugui.datatx.core.executor.JobExecutor;
import com.wugui.datatx.core.handler.IJobHandler;
import com.wugui.datatx.core.log.JobFileAppender;
import com.wugui.datatx.core.log.JobLogger;
import com.wugui.datatx.core.util.ShardingUtil;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * handler thread
 *
 * @author xuxueli 2016-1-16 19:52:47
 */
@Slf4j
public class JobThread extends Thread {

  private final int jobId;
  @Getter private final IJobHandler handler;
  private final LinkedBlockingQueue<TriggerParam> triggerQueue;
  private final Set<Long> triggerLogIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID

  private volatile boolean toStop = false;
  private String stopReason;

  private boolean running = false; // if running job
  private int idleTimes = 0; // idel times

  public JobThread(int jobId, IJobHandler handler) {
    this.jobId = jobId;
    this.handler = handler;
    this.triggerQueue = new LinkedBlockingQueue<>();
    this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<>());
  }

  /**
   * new trigger to queue
   *
   * @param triggerParam
   * @return
   */
  public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
    // avoid repeat
    if (triggerLogIdSet.contains(triggerParam.getLogId())) {
      log.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
      return new ReturnT<>(
          ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
    }

    triggerLogIdSet.add(triggerParam.getLogId());
    triggerQueue.add(triggerParam);
    return ReturnT.SUCCESS;
  }

  /**
   * kill job thread
   *
   * @param stopReason
   */
  public void toStop(String stopReason) {
    /**
     * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)， 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
     * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式；
     */
    this.toStop = true;
    this.stopReason = stopReason;
  }

  /**
   * is running job
   *
   * @return
   */
  public boolean isRunningOrHasQueue() {
    return running || !triggerQueue.isEmpty();
  }

  @Override
  public void run() {

    // init
    try {
      handler.init();
    } catch (Throwable e) {
      log.error(e.getMessage(), e);
    }

    // execute
    while (!toStop) {
      running = false;
      idleTimes++;

      TriggerParam tgParam = null;
      ReturnT<String> executeResult = null;
      try {
        // to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of
        // poll(timeout)
        tgParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
        if (tgParam != null) {
          running = true;
          idleTimes = 0;
          triggerLogIdSet.remove(tgParam.getLogId());

          // log filename, like "logPath/yyyy-MM-dd/9999.log"
          String logFileName =
              JobFileAppender.makeLogFileName(
                  new Date(tgParam.getLogDateTime()), tgParam.getLogId());
          JobFileAppender.CONTEXT_HOLDER.set(logFileName);
          ShardingUtil.setShardingVo(
              new ShardingUtil.ShardingVO(
                  tgParam.getBroadcastIndex(), tgParam.getBroadcastTotal()));

          // execute
          JobLogger.log(
              "<br>----------- datax-web job execute start -----------<br>----------- Param:"
                  + tgParam.getExecutorParams());

          if (tgParam.getExecutorTimeout() > 0) {
            // limit timeout
            Thread futureThread = null;
            try {
              final TriggerParam tgParamT = tgParam;
              FutureTask<ReturnT<String>> futureTask =
                  new FutureTask<>(() -> handler.execute(tgParamT));
              futureThread = new Thread(futureTask);
              futureThread.start();

              executeResult = futureTask.get(tgParam.getExecutorTimeout(), TimeUnit.MINUTES);
            } catch (TimeoutException e) {

              JobLogger.log("<br>----------- datax-web job execute timeout");
              JobLogger.log(e);

              executeResult =
                  new ReturnT<>(IJobHandler.FAIL_TIMEOUT.getCode(), "job execute timeout ");
            } finally {
              futureThread.interrupt();
            }
          } else {
            // just execute
            executeResult = handler.execute(tgParam);
          }

          if (executeResult == null) {
            executeResult = IJobHandler.FAIL;
          } else {
            executeResult.setMsg(
                (executeResult != null
                        && executeResult.getMsg() != null
                        && executeResult.getMsg().length() > 50000)
                    ? executeResult.getMsg().substring(0, 50000).concat("...")
                    : executeResult.getMsg());
            executeResult.setContent(null); // limit obj size
          }
          JobLogger.log(
              "<br>----------- datax-web job execute end(finish) -----------<br>----------- ReturnT:"
                  + executeResult);

        } else {
          if (idleTimes > 30) {
            if (triggerQueue.isEmpty()) { // avoid concurrent trigger causes jobId-lost
              JobExecutor.removeJobThread(jobId, "executor idel times over limit.");
            }
          }
        }
      } catch (Throwable e) {
        if (toStop) {
          JobLogger.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
        }

        StringWriter stringWriter = new StringWriter();
        e.printStackTrace(new PrintWriter(stringWriter));
        String errorMsg = stringWriter.toString();
        executeResult = new ReturnT<>(ReturnT.FAIL_CODE, errorMsg);

        JobLogger.log(
            "<br>----------- JobThread Exception:"
                + errorMsg
                + "<br>----------- datax-web job execute end(error) -----------");
      } finally {
        // 终止操作暂不监控状态
        if (tgParam != null && tgParam.getJobId() != -1) {
          // callback handler info
          if (!toStop) {
            // commonm
            TriggerCallbackThread.pushCallBack(
                new HandleCallbackParam(
                    tgParam.getLogId(), tgParam.getLogDateTime(), executeResult));
          } else {
            // is killed
            ReturnT<String> stopResult =
                new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job running, killed]");
            TriggerCallbackThread.pushCallBack(
                new HandleCallbackParam(tgParam.getLogId(), tgParam.getLogDateTime(), stopResult));
          }
        }
      }
    }

    // callback trigger request in queue
    while (triggerQueue != null && !triggerQueue.isEmpty()) {
      TriggerParam triggerParam = triggerQueue.poll();
      if (triggerParam != null) {
        // is killed
        ReturnT<String> stopResult =
            new ReturnT<String>(
                ReturnT.FAIL_CODE, stopReason + " [job not executed, in the job queue, killed.]");
        TriggerCallbackThread.pushCallBack(
            new HandleCallbackParam(
                triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));
      }
    }

    // destroy
    try {
      handler.destroy();
    } catch (Throwable e) {
      log.error(e.getMessage(), e);
    }

    log.info(">>>>>>>>>>> datax-web JobThread stoped, hashCode:{}", Thread.currentThread());
  }
}
